package org.xfs.flume.interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import com.alibaba.fastjson.JSONObject;

public class TimeStampInterceptor implements Interceptor {

	@Override
	public void initialize() {

	}

	@Override
	public Event intercept(Event event) {
		
		//从body取出ts时间戳放入header中的timestamp的key中
		
		//1、获取body中的ts 数据
		byte[]body=event.getBody();
		String log=new String(body,StandardCharsets.UTF_8);
		JSONObject jsonObject=JSONObject.parseObject(log);
		String ts=jsonObject.getString("ts");
		event.getHeaders().put("timestamp",ts);
		return event;
	}

	@Override
	public List<Event> intercept(List<Event> events) {
		for(Event event:events) {
			this.intercept(event);
		}
		return events;
	}

	@Override
	public void close() {

	}

	public static class Builder implements Interceptor.Builder{

		@Override
		public void configure(Context context) {
			
		}

		@Override
		public Interceptor build() {
			return new TimeStampInterceptor();
		}
		
	}
}
